package streaming.time.processingTime;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * 定义处理时间(Processing Time)
 * 创建表的 DDL 中指定
 */
public class ProcessingTime3 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String pathForm = "D:\\IdeaProjects\\springboot-flink-1\\flinkTutorial\\src\\main\\resources\\sensor.txt";
        String fileFormat = "csv";

        StringBuffer sb1 = new StringBuffer();
        sb1.append("CREATE TABLE inputTable (");
        sb1.append("id STRING, ts BIGINT, temp DOUBLE, pt AS PROCTIME()");
        sb1.append(") WITH (");
        sb1.append("'connector.type' = 'filesystem', ");
        sb1.append("'connector.path' = '").append(pathForm).append("', ");
        sb1.append("'format.type' = '").append(fileFormat).append("'");
        sb1.append(")");
        System.out.println("inputTable SQL: " + sb1.toString());
        tableEnv.executeSql(sb1.toString());
        // 转换
        Table inputTable = tableEnv.from("inputTable");
        // 输出结构
        inputTable.printSchema();
        // 打印输出
        tableEnv.toAppendStream(inputTable, Row.class).print();

        env.execute();

    }
}
